package com.xy.config.mqtt;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.xy.api.model.dto.mqtt.MqttDTO;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * mqtt连接、订阅发送服务、消息发送。如果加入回调方法，则等待接收当前msgId的响应消息，成功或失败都会回调响应结果
 */
@Slf4j
public class MessageClient {

    private static final Gson GSON = new GsonBuilder().disableHtmlEscaping().create();

    private final static String SUBSCRIBE_TOPIC_MODULE = "/matrix/%s/#";
    private volatile static MessageClient INSTANCE;

    private final boolean cleanSession;

    /**
     * 等待响应时间
     **/
    private Integer responseTimes;

    public final MqttAsyncClient mqttClient;

    private IListener listener;

    /**
     * 断线重连后，自动订阅
     */
    private final CopyOnWriteArraySet<String> subscribeTopicSet = new CopyOnWriteArraySet<>();

    private MessageClient(String broker, String clientId, boolean cleanSession, Integer responseTimes, MqttDTO MqttDTO) throws MqttException {
        this.cleanSession = cleanSession;
        this.responseTimes = responseTimes;
        mqttClient = new MqttAsyncClient(broker, clientId, new MemoryPersistence());
        mqttClient.setCallback(new IpCourtRoomMqttCallback());
        try {
            MqttConnectOptions options = getMqttConnectOptions(MqttDTO);
            mqttClient.connect(options).waitForCompletion();
        } catch (Exception e) {
            log.error("connect error ", e);
        }
    }

    private MqttConnectOptions getMqttConnectOptions(MqttDTO MqttUserDTO) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(MqttUserDTO.getUsername());
        options.setPassword(MqttUserDTO.getPassword().toCharArray());
        options.setCleanSession(cleanSession);
        options.setAutomaticReconnect(true);
        options.setKeepAliveInterval(60);
        options.setMaxInflight(1000);
        options.setConnectionTimeout(3);
        options.setMaxReconnectDelay(5000);
        return options;
    }

    public static synchronized void init(final String broker, final String clientId, boolean cleanSession, Integer responseTimes, MqttDTO mqttDTO) {
        log.info("init mqtt broker:{},clientId:{}", broker, clientId);
        if (INSTANCE != null) {
            return;
        }
        try {
            INSTANCE = new MessageClient(broker, clientId, cleanSession, responseTimes, mqttDTO);
        } catch (Exception ex) {
            log.error("init mqtt error", ex);
        }
    }

    public static MessageClient getInstance() {
        return INSTANCE;
    }

    public MessageClient setListener(IListener listener) {
        this.listener = listener;
        return this;
    }

    class IpCourtRoomMqttCallback implements MqttCallbackExtended {
        @Override
        public void connectComplete(boolean b, String s) {
            log.info("connectComplete clientId:{}", mqttClient.getClientId());
            subscribeTopicSet.add(String.format(SUBSCRIBE_TOPIC_MODULE, mqttClient.getClientId()));
            for (String topic : subscribeTopicSet) {
                doSubscribe(topic);
            }
            if (listener == null) {
                return;
            }
            listener.onConnected();
        }

        @Override
        public void connectionLost(Throwable throwable) {
            log.error("connectionLost clientId:{}", mqttClient.getClientId());
            if (listener == null) {
                return;
            }
            listener.onError(throwable);
        }

        @Override
        public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
            String payload = new String(mqttMessage.getPayload(), StandardCharsets.UTF_8);
            try {
                //content内容json字符串，否则转换失败
                final MessageModel messageModel = GSON.fromJson(payload, MessageModel.class);
                if (Objects.isNull(messageModel) || Objects.equals(messageModel.getFrom(), mqttClient.getClientId())) {
                    return;
                }
                log.debug("message arrived topic:{} type:{}", topic, messageModel.getType());
            } catch (Exception e) {
                log.error("messageArrived error topic:{} payload:{}", topic, payload, e);
            }
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

        }

        public void doSubscribe(String topic) {
            doSubscribe(topic, 2);
        }

        private void doSubscribe(String topic, int qos) {
            subscribeTopicSet.add(topic);
            try {
                mqttClient.subscribe(topic, qos);
            } catch (MqttException e) {
                log.error("doSubscribe error " + topic + "---" + mqttClient.getClientId(), e);
            }
        }
    }
}
